package com.platform.apex.output;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Queue;
import java.util.concurrent.ArrayBlockingQueue;

import com.datatorrent.api.Context.OperatorContext;
import com.datatorrent.api.DefaultInputPort;
import com.datatorrent.common.util.BaseOperator;
import com.platform.apex.input.MyJdbcStore;

import scala.Char;

public class MySqlOutputRealTimeOperator extends BaseOperator {

	protected transient MyJdbcStore store;
	private String databaseUrl;
	private String userName;
	private String password;
	private String databaseDriver;
	private String tableName;
	protected transient Queue<Object> tupleBatch;
	private List<String> tableList;

	private Map<String, Boolean> isCreateMap = new HashMap<>();

	@Override
	public void setup(OperatorContext context) {
		// TODO Auto-generated method stub
		super.setup(context);

		tupleBatch = new ArrayBlockingQueue<Object>(100);
		databaseUrl = databaseUrl + "?useUnicode=true&characterEncoding=UTF-8";
		store = new MyJdbcStore(databaseUrl, userName, password, databaseDriver);
		tableList = Arrays.asList(tableName.split(","));

		// MetaDataUtil.initDBConnector(connectionStr, jdbcDriver, userName,
		// password);
		// try {
		// MySqlManager.init(MetaDataUtil.getMetadataConnector());
		// } catch (Exception e) {
		// // TODO Auto-generated catch block
		// e.printStackTrace();
		// }
	}

	public final transient DefaultInputPort<Object> input = new DefaultInputPort<Object>() {
		@Override
		@SuppressWarnings("UseOfSystemOutOrSystemErr")
		public void process(Object tuple) {
			tupleBatch.add(tuple);

			if (tupleBatch.size() >= 100) {
				try {
					batchInsert();
				} catch (Exception e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
			}
		}
	};

	@Override
	public void beginWindow(long windowId) {
		// TODO Auto-generated method stub
		super.beginWindow(windowId);
	}

	@Override
	public void endWindow() {
		// TODO Auto-generated method stub
		super.endWindow();
		try {
			batchInsert();
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	public String getDatabaseUrl() {
		return databaseUrl;
	}

	public void setDatabaseUrl(String databaseUrl) {
		this.databaseUrl = databaseUrl;
	}

	public String getUserName() {
		return userName;
	}

	public void setUserName(String userName) {
		this.userName = userName;
	}

	public String getPassword() {
		return password;
	}

	public void setPassword(String password) {
		this.password = password;
	}

	public String getDatabaseDriver() {
		return databaseDriver;
	}

	public void setDatabaseDriver(String databaseDriver) {
		this.databaseDriver = databaseDriver;
	}

	public void batchInsert() throws Exception {

		Map<String, List<Map<String, Object>>> map = new HashMap<>();
		for (String s : tableList) {
			map.put(s, new ArrayList<Map<String, Object>>());
		}
		while (!tupleBatch.isEmpty()) {
			Object tuple = tupleBatch.remove();

			Map<String, Object> mapSingle = (Map<String, Object>) tuple;

			String tableName = (String) mapSingle.remove("tableName");
			List<Map<String, Object>> list = map.get(tableName);
			list.add(mapSingle);
		}

		insert(map);
		// MySqlManager.create(list);
	}

	public String getTableName() {
		return tableName;
	}

	public void setTableName(String tableName) {
		this.tableName = tableName;
	}

	public void insert(Map<String, List<Map<String, Object>>> map) {

		Connection connInsert = null;
		PreparedStatement prest = null;

		try {
			store.connect();
			connInsert = store.getConnection();

			// 可以得到有多少列
			connInsert.setAutoCommit(false);

			for (Entry<String, List<Map<String, Object>>> en : map.entrySet()) {
				String tableName = en.getKey();

				List<Map<String, Object>> mapList = en.getValue();

				StringBuilder sbCloumn = new StringBuilder();
				StringBuilder sbValues = new StringBuilder();
				sbCloumn.append("insert into " + tableName + " (");
				sbValues.append(" VALUES(");
				if (mapList.size() > 0) {
					Map<String, Object> mapSingleOne = mapList.get(0);
					// 拼装sql语句

					for (Entry<String, Object> enSingle : mapSingleOne.entrySet()) {
						sbCloumn.append(enSingle.getKey() + ",");
						sbValues.append("?,");
					}
					if (isCreateMap.get(tableName) == null) {
						createTable(tableName, mapSingleOne);
					}
					String sqlCloumn = sbCloumn.substring(0, sbCloumn.length() - 1);
					String sqlValues = sbValues.substring(0, sbValues.length() - 1);

					String insertSql = sqlCloumn + ")" + sqlValues + ")";

					prest = connInsert.prepareStatement(insertSql, ResultSet.TYPE_SCROLL_SENSITIVE,
							ResultSet.CONCUR_READ_ONLY);
					for (Map<String, Object> mapSingle : mapList) {
						int count = 0;
						Object[] objects = new Object[mapSingle.size()];
						for (Entry<String, Object> ee : mapSingle.entrySet()) {
							objects[count] = ee.getValue();
							if (objects[count] == null) {
								prest.setString(count + 1, null);
							} else if (objects[count] instanceof String) {
								prest.setString(count + 1, (String) objects[count]);
							} else if (objects[count] instanceof Integer) {
								prest.setInt(count + 1, (int) objects[count]);
							} else if (objects[count] instanceof Boolean) {
								prest.setBoolean(count + 1, (boolean) objects[count]);
							} else if (objects[count] instanceof java.sql.Date) {
								prest.setDate(count + 1, (java.sql.Date) objects[count]);
							} else if (objects[count] instanceof java.lang.Character) {
								prest.setObject(count + 1, objects[count]);
							}else if(objects[count] instanceof java.sql.Timestamp){
								prest.setTimestamp(count + 1, (java.sql.Timestamp) objects[count]);
							}
							count++;
						}
						prest.addBatch();
					}
					prest.executeBatch();
					connInsert.commit();
				}
			}
			connInsert.close();
		} catch (Exception e) {
			// TODO: handle exception
			e.printStackTrace();
		}

	}

	/**
	 * adjust_flow attact_content attact_qty back_opinion back_opt_id
	 * back_opt_name back_opt_time back_status belong_work close_opinion
	 * close_opt_id close_opt_name close_opt_time create_dept_id
	 * create_dept_name create_opt_id create_opt_name create_opt_time
	 * data_resource data_type deal_opinion deal_opt_id deal_opt_name
	 * deal_opt_time deal_part_id deal_part_name deal_result deal_start_time
	 * deal_status deal_time deal_view detail_id dispatch_opt_id
	 * dispatch_opt_name flow_id is_close is_valid is_warn limit_time next_dept
	 * next_person node_id opt_id opt_name opt_order opt_time over_status
	 * pre_tran_id receive_opinion receive_opt_id receive_opt_name
	 * receive_opt_time receive_status remark step_id supervise_id
	 * supervise_level supervise_opinion supervise_qty supervise_title
	 * supervise_type timely_status timely_time tran_id tran_title urge_id
	 * urge_opinion urge_qty urge_title warn_content warn_id warn_qty warn_sort
	 * warn_title warn_type wish_opt_time work_id work_status
	 * 
	 * @param tableName
	 */
	public void createTable(String tableName, Map<String, Object> map) {
		System.out.println(map.size());
		Connection conn = null;
		Statement stmt = null;
		try {
			store.connect();
			conn = store.getConnection();

			stmt = conn.createStatement();

			StringBuilder sb = new StringBuilder();
			sb.append("CREATE TABLE " + tableName + " (");
			for (Entry<String, Object> en : map.entrySet()) {
				
				
				if (en.getValue() instanceof Integer) {
					sb.append(en.getKey() + " ").append("INT(10),");
				} else if (en.getValue() instanceof Boolean) {
					sb.append(en.getKey() + " ").append("TINYINT(2),");
				} else if (en.getValue() instanceof java.sql.Date) {
					sb.append(en.getKey() + " ").append("DATE,");
				} else if (en.getValue() instanceof Double) {
					sb.append(en.getKey() + " ").append("DOUBLE,");
				} else if (en.getValue() instanceof Long) {
					sb.append(en.getKey() + " ").append("BIGINT,");
				} else if (en.getValue() instanceof Float) {
					sb.append(en.getKey() + " ").append("FLOAT,");
				} else if (en.getValue() instanceof Float) {
					sb.append(en.getKey() + " ").append("FLOAT,");
				} else if (en.getValue() instanceof java.sql.Timestamp) {
					sb.append(en.getKey() + " ").append("DATETIME,");
				} else if (en.getValue() == null) {
					sb.append(en.getKey() + " ").append("VARCHAR(255),");
				} else if (en.getValue() instanceof java.lang.Character) {
					sb.append(en.getKey() + " ").append("CHAR(20),");
				} else if (en.getValue() instanceof String) {
					sb.append(en.getKey() + " ").append("VARCHAR(255),");
				}
			}

			String creatSql = sb.substring(0, sb.length() - 1);
			creatSql = creatSql + ")";

			System.out.println("creatSql" + creatSql);
			stmt.executeUpdate(creatSql);
			isCreateMap.put(tableName, true);
			System.out.println("Created table in given database...");
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			try {
				if (stmt != null)
					conn.close();
			} catch (SQLException se) {
			}
			try {
				if (conn != null)
					conn.close();
			} catch (SQLException se) {
				se.printStackTrace();
			} // end finally try
		} // end try
	}

}
